package com.zhu.function;


import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.zhu.utils.DIMUtils;
import com.zhu.utils.ZhuDruidDataSourceUtil;
import com.zhu.utils.ZhuThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.lang.reflect.InvocationTargetException;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *
 * DIM  维度表层  异步IO函数
 * 查询Hbase维表  关联字段
 * 传入参数 可以通过反射 获取要关联的字段  本类不适用此方法
 * 模板方法设计模式 定义抽象类
 *
 * 异常处理 维表关联重要程度
 *
 */
public abstract class DIMAsyncFunction<T> extends RichAsyncFunction<T,T> implements DIMJoinFunction<T> {

    private DruidDataSource dataSource = null;
    private ThreadPoolExecutor threadPoolExecutor = null;
    private String tableName;
    //关联字段

    public DIMAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //Druid  ThreadPool
        dataSource = ZhuDruidDataSourceUtil.createDataSource();
        threadPoolExecutor = ZhuThreadPoolUtil.getThreadPoolExecutor();   //单例
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //获取连接
                try {
                    DruidPooledConnection connection = dataSource.getConnection();
                    //select dim get data
                    String key = getKey(input);
                    JSONObject dimInfo = DIMUtils.getDimInfo(connection, tableName, key);

                    //补充维表数据
                    if(dimInfo != null){

                        join(input,dimInfo);
                    }
                    connection.close();  //归还连接池连接
                    //写出结果
                    resultFuture.complete(Collections.singletonList(input));
                } catch (SQLException | InvocationTargetException e) {
                    e.printStackTrace();
                    System.out.println("join dim tables from Hbase false " + input +" connect" + tableName);
                    //throw new RuntimeException();
                }
            }
        });
    }

    /**
     * 关联维表
     * @param input 输出数据
     * @param dimInfo  维表数据
     */
   public abstract void join(T input, JSONObject dimInfo);

    /**
     * 获取主键
     * @param input  输出数据
     * @return
     */
    public abstract String getKey(T input);


    /**
     * 数据连接超时
     * @param input
     * @param resultFuture
     * @throws Exception
     */
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("timeout>>>" + input);
    }
}
